package com.simon.flow;

import java.util.concurrent.Flow;

public class TwoSubscriber implements Flow.Subscriber<String> {

    private static final String SEVEN = "7";
    
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        System.out.println(">>> start...");
        this.subscription = subscription;

        this.subscription.request(1);
    }

    @Override
    public void onNext(String item) {
        System.out.println(">>> received... item: " + item);

        if (SEVEN.equals(item)) {
            this.subscription.cancel();
            return;
        }
        this.subscription.request(1);

    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println(">>> error...");
    }

    @Override
    public void onComplete() {
        System.out.println(">>> complete...");
    }
}
